package com.qizhidao.j2cache.redis;

/*
 * Copyright 2011-2021 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import net.oschina.j2cache.CacheChannel;
import net.oschina.j2cache.CacheObject;
import org.springframework.data.redis.connection.BitFieldSubCommands;
import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.lang.Nullable;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


public class J2CacheValueOperations<K, V> extends AbstractJOperations<K, V> implements ValueOperations<K, V> {

    private final static String REDIS_SEPARATOR = ":";
    private CacheChannel channel;

    protected String region;

    public J2CacheValueOperations(RedisTemplate<K, V> template, CacheChannel channel, String region) {
        super(template);
        this.channel = channel;
        this.region = region;
    }

    public boolean delete(String... key) {
        String[] redisKey = handleKey(region, key);
        channel.evict(region, redisKey);
        return true;
    }

    @Override
    public V get(Object key) {
        String redisKey = handleKey(region, (String) key);
        return (V) channel.get(region, redisKey).getValue();
    }


    public V get(String region, Object key) {
        String redisKey = handleKey(region, (String) key);
        return (V) channel.get(region, redisKey).getValue();
    }


    @Override
    public V getAndSet(K key, V newValue) {
        V oldValue = this.get(key);
        this.set(key, newValue);
        return oldValue;
    }

    public V getAndSet(K key, V newValue, long timeout, TimeUnit timeUnit) {
        V oldValue = this.get(key);
        this.set(key, newValue, timeout, timeUnit);
        return oldValue;
    }

    @Override
    public Long increment(K key) {
        byte[] rawKey = rawKey(key);
        return execute(connection -> connection.incr(rawKey), true);
    }

    @Override
    public Long increment(K key, long delta) {

        byte[] rawKey = rawKey(key);
        return execute(connection -> connection.incrBy(rawKey, delta), true);
    }

    @Override
    public Double increment(K key, double delta) {

        byte[] rawKey = rawKey(key);
        return execute(connection -> connection.incrBy(rawKey, delta), true);
    }

    @Override
    public Long decrement(K key) {

        byte[] rawKey = rawKey(key);
        return execute(connection -> connection.decr(rawKey), true);
    }

    @Override
    public Long decrement(K key, long delta) {

        byte[] rawKey = rawKey(key);
        return execute(connection -> connection.decrBy(rawKey, delta), true);
    }

    @Override
    public Integer append(K key, String value) {
        String oldValue = (String) this.get(key);
        StringBuilder stringBuilder = new StringBuilder(oldValue);
        stringBuilder.append(value);
        this.set(key, (V) stringBuilder.toString());
        return stringBuilder.length();
    }


    @Override
    public String get(K key, long start, long end) {
        String redisKey = handleKey(region, (String) key);
        String value = (String) channel.get(region, redisKey).getValue();
        if (!StringUtils.isEmpty(value)) {
            if (end >= value.length()) {
                end = value.length() - 1;
            }
            return value.substring((int) start, (int) end + 1);
        }
        return "";
    }

    @Override
    public List<V> multiGet(Collection<K> keys) {
        if (keys.isEmpty()) {
            return Collections.emptyList();
        }
        List<V> result = new ArrayList<>();
        List<String> j2CacheKeys = keys.stream().map(e -> {
            return handleKey(region, (String) e);
        }).collect(Collectors.toList());
        Map<String, CacheObject> map = channel.get(region, j2CacheKeys);
        Iterator<Map.Entry<String, CacheObject>> iterator = map.entrySet().iterator();
        keys.forEach(e -> {
            boolean flag = false;
            while (iterator.hasNext()) {
                Map.Entry<String, CacheObject> entry = iterator.next();
                if (Objects.equals(e, entry.getKey())) {
                    result.add((V) entry.getValue().getValue());
                    flag = true;
                    break;
                }
            }
            if (!flag) {
                result.add(null);
            }
        });
        return result;
    }

    @Override
    public void multiSet(Map<? extends K, ? extends V> m) {

        if (m.isEmpty()) {
            return;
        }
        Map<String, Object> param = new TreeMap<>();
        Iterator<? extends Map.Entry<? extends K, ? extends V>> iterator = m.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry entry = iterator.next();
            String key = handleKey(region, (String) entry.getKey());
            param.put(key, entry.getValue());
        }
        channel.set(region, param, 1000);
    }

    public void multiSet(Map<? extends K, ? extends V> m, long timeout, TimeUnit timeUnit) {
        if (m.isEmpty()) {
            return;
        }
        Map<String, Object> param = new TreeMap<>();
        Iterator<? extends Map.Entry<? extends K, ? extends V>> iterator = m.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry entry = iterator.next();
            String key = handleKey(region, (String) entry.getKey());
            param.put(key, entry.getValue());
        }
        channel.set(region, param, timeUnit.toSeconds(timeout));
    }

    @Override
    public Boolean multiSetIfAbsent(Map<? extends K, ? extends V> m) {

        if (m.isEmpty()) {
            return true;
        }

        Map<byte[], byte[]> rawKeys = new LinkedHashMap<>(m.size());

        for (Map.Entry<? extends K, ? extends V> entry : m.entrySet()) {
            rawKeys.put(rawKey(entry.getKey()), rawValue(entry.getValue()));
        }

        return execute(connection -> connection.mSetNX(rawKeys), true);
    }

    @Override
    public void set(K key, V value) {
        String redisKey = handleKey(region, (String) key);
        channel.set(region, redisKey, value);
    }

    public void set(String region, K key, V value) {
        String redisKey = handleKey(region, (String) key);
        channel.set(region, redisKey, value);
    }

    @Override
    public void set(K key, V value, long timeout, TimeUnit unit) {
        String redisKey = handleKey(region, (String) key);
        channel.set(region, redisKey, value, unit.toSeconds(timeout));
    }

    /**
     * 只有不存在KEY才设置值，否则无效
     *
     * @param key
     * @param value
     * @return
     */
    @Override
    public Boolean setIfAbsent(K key, V value) {

        byte[] rawKey = rawKey(key);
        byte[] rawValue = rawValue(value);
        return execute(connection -> connection.setNX(rawKey, rawValue), true);
    }

    @Override
    public Boolean setIfAbsent(K key, V value, long timeout, TimeUnit unit) {

        byte[] rawKey = rawKey(key);
        byte[] rawValue = rawValue(value);

        Expiration expiration = Expiration.from(timeout, unit);
        return execute(connection -> connection.set(rawKey, rawValue, expiration, SetOption.ifAbsent()), true);
    }

    /**
     * 如果存在key才设置值，否则无效
     *
     * @param key
     * @param value
     * @return
     */
    @Nullable
    @Override
    public Boolean setIfPresent(K key, V value) {

        byte[] rawKey = rawKey(key);
        byte[] rawValue = rawValue(value);

        return execute(connection -> connection.set(rawKey, rawValue, Expiration.persistent(), SetOption.ifPresent()), true);
    }

    @Nullable
    @Override
    public Boolean setIfPresent(K key, V value, long timeout, TimeUnit unit) {

        byte[] rawKey = rawKey(key);
        byte[] rawValue = rawValue(value);

        Expiration expiration = Expiration.from(timeout, unit);
        return execute(connection -> connection.set(rawKey, rawValue, expiration, SetOption.ifPresent()), true);
    }

    @Override
    public void set(K key, V value, long offset) {
        String oldValue = (String) this.get(key);
        if (StringUtils.isEmpty(oldValue) || oldValue.length() < 6) {
            this.append(key, (String) value);
            return;
        }
        StringBuilder stringBuilder = new StringBuilder(oldValue.substring(0, (int) offset));
        stringBuilder.append(value);
        this.set(key, (V) stringBuilder.toString());
    }

    @Override
    public Long size(K key) {
        String redisKey = handleKey(region, (String) key);
        String val = (String) this.get(redisKey);
        if (StringUtils.isEmpty(val)) {
            return null;
        }
        return Long.valueOf(val.length());
    }

    @Override
    public Boolean setBit(K key, long offset, boolean value) {

        byte[] rawKey = rawKey(key);
        return execute(connection -> connection.setBit(rawKey, offset, value), true);
    }

    @Override
    public Boolean getBit(K key, long offset) {

        byte[] rawKey = rawKey(key);
        return execute(connection -> connection.getBit(rawKey, offset), true);
    }

    @Override
    public List<Long> bitField(K key, final BitFieldSubCommands subCommands) {

        byte[] rawKey = rawKey(key);
        return execute(connection -> connection.bitField(rawKey, subCommands), true);
    }

    /**
     * 处理key，兼容j2cache的region
     *
     * @param region
     * @param key
     * @return
     */
    private String handleKey(String region, String key) {
        if (key.startsWith(region + REDIS_SEPARATOR)) {
            return key.replace(region + REDIS_SEPARATOR, "");
        }
        return key;
    }

    private String[] handleKey(String region, String... key) {
        for (int i = 0; i < key.length; i++) {
            if (key[i].startsWith(region + REDIS_SEPARATOR)) {
                key[i] = key[i].replace(region + REDIS_SEPARATOR, "");
            }
        }
        return key;
    }
}
